from pyspark.streaming import StreamingContext
from pyspark import SparkConf,SparkContext

sf = SparkConf().setMaster("spark://192.168.0.104:7077").setAppName("xxxxxx")
sc = SparkContext(conf=sf)
sc.setLogLevel("ERROR")
s = StreamingContext(sc, 1)
s.checkpoint("file:///media/psf/Home/Workspace/Rimi/P1901/lessons/spark/checkpoints")
s.textFileStream()
dstream = s.socketTextStream("192.168.0.103", 7777)
dstream.window(3)
dstream = dstream.flatMap(lambda x: x.split()).map(lambda x: (x, 1)).groupByKey().mapValues(lambda x: len([w for w in x]))
dstream.pprint()
s.start()
s.awaitTermination()